#include <carbon/core/queued_thread_pool.h>

CARBON_NS_BEGIN


bool QueuedThreadPool::start(carbon::Error * err)
{
    assert(m_min_threads > 0 || m_max_threads > 0); // 未初始化，初始化后必然有一个不是零。

    for ( int i = 0; i < m_min_threads; ++i ) {
        bool ok = this->startThread(); 
        assert(ok);
    }
    return true;
}

void QueuedThreadPool::stop()
{
    // 一个空任务，用于唤醒所有可能空闲等待的线程
    auto noop = [](){ };
    
    this->m_need_stop = true;   // 标记当前线程池需要停止

    // 向任务队列发送空任务，唤醒所有线程
    int threads = m_thread_count.load();
    for(int i = 0; i < threads && m_tasks.offer(noop); ++i) {}

    // 等待线程退出
    this->joinThreads();

    // 清理任务队列
    while ( auto opt = m_tasks.poll() ) ;
}


/// 发送一个可执行任务到线程池的任务队列。非阻塞。
/// 如果队列中任务积压量过大，且线程数未到最大值，启动一个新线程
bool QueuedThreadPool::execute(carbon::RunnableFn fn, carbon::Error * err)
{
    // 任务写入队列
    bool ok = m_tasks.offer(std::move(fn));
    if ( ok ) return true;  // 写入完成，返回

    // 写入失败，则尝试创建线程。创建线程后，由于线程创建需要时间，立即重新尝试写入任务，大概率仍然失败，因此干脆不做重试。
    this->startThread();

    return false;   // 当前任务写入失败。
}

bool QueuedThreadPool::execute(RunnableFn fn, std::chrono::milliseconds timeout, carbon::Error * err)
{
    // 任务写入队列
    bool ok = m_tasks.offer(std::move(fn));
    if ( ok ) return true;  // 写入完成，返回

    // 写入失败，则尝试创建线程。
    this->startThread();

    // 尝试写入，或等待超时
    return m_tasks.offer(std::move(fn), timeout);
}

void QueuedThreadPool::joinThreads()
{
    // 等待线程停止并回收线程资源
    for ( size_t i = m_threads.size(); i > 0; --i) {
        auto & th = m_threads[i-1];
        if ( th.joinable() ) th.join();
    }
}

bool QueuedThreadPool::startThread()
{
    int thread_index = -1;
    int thread_count = 0;

    do { 
        thread_count = m_thread_count.load();
        if ( thread_count >= m_max_threads ) {
            fprintf(stderr, "QueuedThreadPool::startThread thread full, threads: %d\n", thread_count);
            break; // 线程数已满
        }
        
        thread_index = thread_count;    // 获取线程在表中索引号
    } while( !m_thread_count.compare_exchange_strong(thread_count, thread_count + 1) );

    if ( thread_index == -1) return false;  // 无需创建线程

    // 在表中指定下标位置创建线程
    auto & thread = m_threads[thread_index];
    if ( thread.joinable() ) thread.join();  // 回收线程资源, 线程主函数退出前，仅递减thread_count，不会对自身做join操作，需要在下次重用前做join，然后重用该thread对象

    std::string tname = this->m_name + "-" + std::to_string(thread_index);
    thread = carbon::Thread([this, thread_index](){ this->threadRun(thread_index); }, tname);

    fprintf(stderr, "QueuedThreadPool::startThread %lx, count: %d\n", thread.id(), m_thread_count.load());
    return true;    // 线程创建成功
}

void QueuedThreadPool::threadRun(int threadIndex)
{
    fprintf(stderr, "QueuedThreadPool::threadRun new thread: %lx\n", pthread_self());
    while( !m_need_stop ) {
        // 循环获取任务，首次等待一定时间，超时或没拿到任务就跳出循环，检查是否需要退出。如果能连续获取到任务，则中间无需检查是否退出。
        for ( auto task = m_tasks.poll(std::chrono::seconds(m_idle_timeout)); task.isPresent() && !m_need_stop; task = m_tasks.poll()) {
            task.get()();   // 执行任务
        }

        if ( m_need_stop ) break;   // 线程需要退出

        // 长时间没有获取到任务，或者只要有一次没拿到任务，就需要判断当前线程是否需要退出。
        //

        // 本线程属于安全区间，无需退出
        if ( threadIndex < m_min_threads ) continue;  

        // 不是末尾线程，无需退出。本线程池约定线程退出按照由后往前依次退出，以便创建时可以根据thread count找到目标线程位置
        int thread_count = m_thread_count.load();
        if ( threadIndex < thread_count - 1) continue;   // 暂时不用退

        bool ok = m_thread_count.compare_exchange_strong(thread_count, threadIndex);
        if ( ok ) break; // 当前是最后一个线程，需要退出。

        // 如果设置失败，则说明实际thread_count被改变，减少是不可能的，因为上面已经判断当前是最后一个线程，其他线程不会退出。
        // 可能性只有线程增加了，这种情况下，当前已经不是最后一个线程线程，无需退出
    }

    fprintf(stderr, "QueuedThreadPool::threadRun thread exit: %lx\n", pthread_self());

    // 线程退出
}


CARBON_NS_END
